package gbench.appdemo.priceline.app;

import org.junit.jupiter.api.Test;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import static gbench.common.tree.LittleTree.IRecord.*;
import static gbench.common.tree.LittleTree.*;
import gbench.common.tree.LittleTree.IRecord;
import gbench.commonApp.data.DataApp;

public class JunitDistruptor {
    
    public class MyData extends DataApp{
        public List<IRecord> sqlquery(final String sql) {
            cfg.set("url", "jdbc:mysql://localhost:3306/futures?serverTimezone=GMT%2B8"); reload();
            return this.jdbc.sql2records(sql);
        }
        
        public IRecord config() {
            cfg.set("url", "jdbc:mysql://localhost:3306/futures?serverTimezone=GMT%2B8"); reload();
            return this.cfg;
        }
    }
    
    @Test
    public void foo() throws InterruptedException {
        final var bufferSize = 16;//环形队列长度，必须是2的N次方
        final var disruptor = new Disruptor<IRecord>(()->REC(),bufferSize,Executors.defaultThreadFactory(),
            ProducerType.MULTI,new YieldingWaitStrategy());
        final var dataApp = new MyData();
        final var latch = new CountDownLatch(2);// 停止信号
        final var ai = new AtomicInteger(0);
        disruptor.handleEventsWith((event,seq,batch)->{
            if(event.bool("close")) {
                latch.countDown();
            }else {
                ai.getAndIncrement();
                System.out.println(MFT("1->event:{0},seq:{1},batch:{2}",event,seq,batch)); 
            }//if
        });
        
        disruptor.handleEventsWith((event,seq,batch)->{
            if(event.bool("close")) {
                latch.countDown();
            }else {
                ai.getAndIncrement();
                System.out.println(MFT("2->event:{0},seq:{1},batch:{2}",event,seq,batch)); 
            }//if
        });
        
        final var p = dataApp.sqlquery("SELECT time, price, volume FROM ohlc1.t_deriv_eg2001 limit 5");
        Executors.newFixedThreadPool(1).execute(()->{// 使用线程异步执行
            Stream.iterate(0, i->i+1).limit(2).parallel().map(i->p).forEach(rr->{
                rr.forEach(rec->{
                    //System.out.println(rec);
                    disruptor.getRingBuffer().publishEvent((r,seq)->r.add(rec).add("seq",seq));
                });
            });
        });
        
        // 发送结束消息
        disruptor.getRingBuffer().publishEvent((r,seq)->r.set("close", true));
        disruptor.start();
        
        // 等待运行结束
        latch.await();
        disruptor.shutdown();
        System.out.println(MFT("{0}",ai));
    }
}
